package com.citusdata.elven.httpRpc;

import com.citusdata.elven.httpRpc.HttpRpcRequestHandler;
import com.citusdata.elven.util.ConfigManager;
import com.google.protobuf.Service;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.commons.configuration.Configuration;
import org.apache.http.*;
import org.apache.http.impl.*;
import org.apache.http.params.*;
import org.apache.http.protocol.*;
import org.apache.log4j.Logger;


// Simple embedded Http server that handles incoming Http requests. The server
// either runs as a stand-alone application, and accepts incoming connections in
// the main thread; or it spawns off a listener thread, and executes within the
// spawned thread. The latter mode exists to simplify unit testing.

public class HttpRpcServer implements Runnable {
  private static final Logger logger = Logger.getLogger(HttpRpcServer.class);
  private static final Configuration config = ConfigManager.getConfig();

  private final ExecutorService executor;
  private final int        port;
  private final boolean    persistent;
  private final HttpParams httpParams = new BasicHttpParams();

  private ServerSocket serverSocket;
  private volatile boolean cancelled;

  // 1. serviceRegistry is shared between HttpRpcRequestHandlers, and therefore
  // needs to allow concurrent accesses. 
  // 2. serviceRegistry's declaration uses Generics; since Service is an interface,
  // Object casting does not work. 
  private ConcurrentMap<String, Service> serviceRegistry 
    = new ConcurrentHashMap<String, Service>();

  public void registerService(Service service) {
    String serviceName = service.getDescriptorForType().getName();
    serviceRegistry.putIfAbsent(serviceName, service);

    logger.info("Registered service interface: " + serviceName);
  }


  public HttpRpcServer(ExecutorService executor, int port, boolean persistent) {
    this.executor   = executor;
    this.port       = port;
    this.persistent = persistent;

    // Look up configuration for HttpRpcServer.SocketTimeout (default: 10 seconds),
    // and HttpRpcServer.SocketBufferSize (default: 4 KB).
    int timeout = config.getInt("HttpRpcServer.SocketTimeout", 10*1000);
    int bufferSize = config.getInt("HttpRpcServer.SocketBufferSize", 4*1024);
    
    httpParams.setIntParameter(CoreConnectionPNames.SO_TIMEOUT, timeout);
    httpParams.setIntParameter(CoreConnectionPNames.SOCKET_BUFFER_SIZE, bufferSize);
    httpParams.setBooleanParameter(CoreConnectionPNames.STALE_CONNECTION_CHECK, false);
    httpParams.setBooleanParameter(CoreConnectionPNames.TCP_NODELAY, true);
  }


  private HttpService initHttpService(boolean reuse, HttpParams params) {
    // Set up protocol processor to process Http headers
    BasicHttpProcessor processor = new BasicHttpProcessor();
    processor.addInterceptor(new ResponseContent());
    processor.addInterceptor(new ResponseConnControl());

    // Set up global request handler for RPC messages
    HttpRequestHandlerRegistry registry = new HttpRequestHandlerRegistry();
    registry.register("*", new HttpRpcRequestHandler(serviceRegistry));
    registry.register(HttpRpcUrl.SHUTDOWN_URL, new HttpRpcShutdownHandler(this));
    
    ConnectionReuseStrategy reuseStrategy = null;
    if (reuse) {
      reuseStrategy = new DefaultConnectionReuseStrategy();
    } else {
      reuseStrategy = new NoConnectionReuseStrategy();
    }

    HttpService service = new HttpService(processor, reuseStrategy,
                                          new DefaultHttpResponseFactory());
    service.setParams(params);
    service.setHandlerResolver(registry);

    return service;
  }


  protected synchronized void cancel() {
    cancelled = true;
    try { 
      if (serverSocket != null) {
        serverSocket.close();
      }
    } catch (IOException ignore) {  
      // Ignore exception  
    } finally {
      serverSocket = null;
    }
  }


  public void run() {
    try {
      serverSocket = new ServerSocket(port);
      logger.info("Server listening on port: " + serverSocket.getLocalPort());

      // server socket accept() is a blocking call, and therefore interrupting
      // it is not trivial. An example of "proper termination" is shown in Java
      // Concurrency in Practice, Listing 7.12.
      while (!cancelled) {
        Socket workerSocket = serverSocket.accept();
        logger.debug("Incoming connection from: " + workerSocket.getInetAddress());

        DefaultHttpServerConnection workerConn = new DefaultHttpServerConnection();
        workerConn.bind(workerSocket, httpParams);

        HttpService httpService = initHttpService(persistent, httpParams);
        Runnable task = new HttpRpcWorkerRunnable(httpService, workerConn);
        executor.execute(task);
      }
    } catch (InterruptedIOException ex) {
      logger.error("I/O interruption, shutting down", ex);
    } catch (IOException ex) {
      // Allow thread to exit through cancel
      logger.info("I/O exception, shutting down" + ex);
    } finally {
      logger.info("Shutting down the server");
      cancel();             // Close server socket if not already closed
      executor.shutdown();  // The HttpRpcServer owns the executor, shut it down
    }
  }


  // Inner class that handles Http specific logic for incoming requests. After
  // processing, this class passes requests to Http request handlers for
  // application logic to be applied.

  class HttpRpcWorkerRunnable implements Runnable {
    private final HttpService service;
    private final HttpServerConnection connection;

    public HttpRpcWorkerRunnable(HttpService service, 
                                 HttpServerConnection connection) {
      this.service = service;
      this.connection = connection;
    }

    public void run() {
      try {
        HttpContext context = new BasicHttpContext();

        while (!Thread.currentThread().isInterrupted() && connection.isOpen()) {
          service.handleRequest(connection, context);
        }
               
      } catch (ConnectionClosedException ex) {
        logger.info("Connection closed: " + ex);
      } catch (SocketTimeoutException ex) {
        logger.info("Socked timed out on read: " + ex);
      } catch (IOException ex) {
        logger.warn("I/O error during socket operation: " + ex);
      } catch (HttpException ex) {
        logger.warn("Unrecoverable HTTP protocol violation: " + ex);
      } finally {
        try { 
          // Connection was created in a different thread; do _not_ call close.
          connection.shutdown();
        } catch (IOException ex) {
          logger.warn("I/O error when closing connection: " + ex);
        }
      }
    }
  }

}
